package cn.le.listener;

import cn.le.po.DeviceDataPO;
import cn.le.service.PgService;
import cn.le.vo.KafkaDownVO;
import com.alibaba.fastjson.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;

import static cn.le.util.Constant.DOWN_TOPIC;

@Component
public class PgListener {

    @Autowired
    private PgService pgService;
    private static List<DeviceDataPO> INSERT_LIST = new ArrayList<>(1100);

    @KafkaListener(topics = DOWN_TOPIC)
    public void pgListener(String msg){
        KafkaDownVO downVO = JSONObject.parseObject(msg,KafkaDownVO.class);

        downVO.getDataList().forEach(propertyData -> {
            DeviceDataPO po = new DeviceDataPO();
            po.setDeviceId(downVO.getDeviceId());
            po.setPropertyId(propertyData.getPropertyId());
            po.setData(propertyData.getData());
            po.setCollTime(po.getCollTime());
            po.setCreateAt(new Date());
            INSERT_LIST.add(po);
        });

        //批量写入数据库
        if (INSERT_LIST.size() > 1000){
            jdbcBachInsert(INSERT_LIST);
        }
    }

    public void jdbcBachInsert(List<DeviceDataPO> downVOList){
        if (CollectionUtils.isEmpty(downVOList)){
            return;
        }
        pgService.jdbcBachInsert(downVOList);
        INSERT_LIST = new ArrayList<>(1100);
    }
}
